package com.yujiahao.bigdata.sql

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object SparkSQL01_DataFrame_RDD_DataSet {
  def main(args: Array[String]): Unit = {
    //TODO 1、SparkSQL直接构建无法构建，包是私有的，使用伴生对象伴生类构建builder
    //SparkSQL对SparkCore进行了封装，所以环境也要发生变化

    val spark = SparkSession.builder
      .master("local")
      .appName("Word Count")
      .getOrCreate()

    import spark.implicits._
    // RDD DataFrame DataSet之间的转换
    //构建RDD
    val rdd: RDD[(Int, String, Int)] = spark.sparkContext.makeRDD(List(
      (1, "liudehua", 50), (2, "zhangxueyou", 60), (3, "wangzuxian", 30)))

    //todo 将RDD转换成DataFrame
    val df: DataFrame = rdd.toDF("id", "name", "age")
    println("-----------------测试将RDD转换成DataFrame---------------")
    df.show()

    //todo 将DataFrame转换成RDD
    val rdd1: RDD[Row] = df.rdd
    println("-----------------测试将DataFrame转换成RDD---------------")
    rdd1.collect().mkString(" ")
    //todo 将DataFrame 转换成DataSet但是这里缺少类型，那就先增加一个类型
    val ds: Dataset[User] = df.as[User]
    println("-----------------测试将DataFrame转换成DataSet---------------")
    ds.show()
    //todo 将DataSet转换成DataFrame
    val newDs: DataFrame = ds.toDF()
    println("-----------------测试将DataSet转换成DataFrame---------------")
    newDs.show()

    //todo RDD转换成DataSet,直接转也可以就是结果不太友好，最好要先处理一下RDD将类型加上
    //最好要先处理一下RDD将类型加上
    val rr: RDD[User] = rdd.map(t => User(t._1, t._2, t._3))
    val rDD_DS: Dataset[User] = rr.toDS()
    //直接转也可以就是结果不太友好
    val value: Dataset[(Int, String, Int)] = rdd.toDS()
    println("-----------------测试将RDD转换成DataSet---------------")
    rDD_DS.show()
    value.show()
    //todo DataSet转换成RDD
    val newRDD: RDD[User] = rDD_DS.rdd
    println("-----------------测试将DataSet转换成RD---------------")
    newRDD.collect().foreach(println)
    //TODO 2、关闭资源
    spark.stop()
  }

  case class User(id: Int, name: String, age: Int)
}
